package gokv

import (
	"gitee.com/Cauchy_AQ/gokv/data"
	"gitee.com/Cauchy_AQ/gokv/fio"
	"gitee.com/Cauchy_AQ/gokv/index"
	"gitee.com/Cauchy_AQ/gokv/utils"
	"io"
	"os"
	"path"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
)

const (
	mergeDirName     = "-merge"
	mergeFinishedKey = "merge.finished"
)

// Merge 清理无效数据，生成 Hint 文件
func (d *DB) Merge() error {
	// 数据库为空
	if d.activeFile == nil {
		return nil
	}

	// 同时只能存在一个线程 Merge
	d.mtx.Lock()
	if d.isMerging {
		d.mtx.Unlock()
		return ErrMergeIsProgress
	}
	d.isMerging = true
	defer func() {
		d.isMerging = false
	}()

	// 查看当前 merge 的数量是否达到的总阈值
	var totalSize int64
	// 如果存在上一次 merge 目录，totalSize 应该表示为 merge 目录的有效数据大小加上原目录未 merge 的所有文件数据大小
	mergePath := d.getMergeDirPath()
	if _, err := os.Stat(mergePath); os.IsNotExist(err) {
		totalSize, err = utils.DirSize(d.opt.DirPath)
		if err != nil {
			d.mtx.Unlock()
			return err
		}
	} else {
		// merge 目录的有效数据
		mergeValidDataSize, err := utils.DataFileDirSize(mergePath, nil)
		if err != nil {
			d.mtx.Unlock()
			return err
		}
		// 获取最近未参与 merge 的文件 id
		fileId, err := d.getNonMergeFileId(mergePath)
		if err != nil {
			d.mtx.Unlock()
			return err
		}
		// 原目录未 merge 的新数据
		afterMergeAllDataSize, err := utils.DataFileDirSize(d.opt.DirPath, func(fileName string) bool {
			words := strings.Split(fileName, ".")
			fid, _ := strconv.Atoi(words[0])
			if uint32(fid) < fileId {
				return false
			}
			return true
		})
		if err != nil {
			d.mtx.Unlock()
			return err
		}
		totalSize = mergeValidDataSize + afterMergeAllDataSize
	}

	if totalSize == 0 {
		d.mtx.Unlock()
		return nil
	}

	// 无效数据占比（无效/(无效+有效)）较低，小于用户设定的阈值比例
	if float32(d.invalidDataSize)/float32(totalSize) < d.opt.DataFileMergeRatio {
		d.mtx.Unlock()
		return ErrMergeRatioUnreached
	}

	// 查看当前剩余空间是否可以容纳 merge 之后的数据量
	available, err := utils.AvailableDiskSize(d.opt.DirPath)
	if err != nil {
		d.mtx.Unlock()
		return err
	}
	// 这里由于还有其他 merge 后产生的文件，未考虑在内，不够精确
	if uint64(totalSize-d.invalidDataSize) >= available {
		d.mtx.Unlock()
		return ErrNoEnoughSpaceForMerge
	}

	// 持久化当前活跃文件
	if err := d.activeFile.Sync(); err != nil {
		d.mtx.Unlock()
		return err
	}
	// 活跃文件转旧的数据文件
	d.olderFiles[d.activeFile.FileId] = d.activeFile
	// 打开新的活跃文件
	if err := d.setActiveFile(); err != nil {
		d.mtx.Unlock()
		return err
	}

	// 记录最近没有参与 merge 的文件 id
	nonMergeFileId := d.activeFile.FileId

	// 取出所有需要 merge 的文件
	// 释放锁，不影响后续用户写入数据到新的数据文件
	var idx = 0
	mergeFiles := make([]*data.DataFile, len(d.olderFiles))
	for _, file := range d.olderFiles {
		mergeFiles[idx] = file
		idx++
	}
	// 清空无效数据量大小
	d.invalidDataSize = 0
	// 释放锁
	d.mtx.Unlock()

	// 从小到大依此 merge
	sort.Slice(mergeFiles, func(i, j int) bool {
		return mergeFiles[i].FileId < mergeFiles[j].FileId
	})

	mergeDirPath := d.getMergeDirPath()
	// 如果merge的目录存在，说明发生过merge，删目录
	if _, err := os.Stat(mergeDirPath); err == nil {
		if err := os.RemoveAll(mergeDirPath); err != nil {
			return err
		}
	}
	// 新建一个 merge 目录
	if err := os.MkdirAll(mergeDirPath, os.ModePerm); err != nil {
		return err
	}

	// 打开一个新的临时 DB 实例
	var tmpOpt Options
	tmpOpt = *d.opt
	var mergeOptions = tmpOpt // 不能修改原配置
	mergeOptions.DirPath = mergeDirPath
	mergeOptions.SyncWrites = false // 保证效率
	mergeDB, err := Open(&mergeOptions)
	defer mergeDB.Close()
	if err != nil {
		return err
	}

	// 打开 Hint 文件存储索引，需要写入数据，使用标准文件 IO
	hintFile, err := data.OpenHintFile(mergeDirPath, fio.StandardFIO)
	defer hintFile.Close()
	if err != nil {
		return err
	}
	// 遍历处理每个数据文件
	for _, dataFile := range mergeFiles {
		var offset int64 = 0
		for {
			logRecord, size, err := dataFile.ReadLogRecord(offset)
			if err != nil {
				if err == io.EOF {
					break
				}
				return err
			}

			realKey, _ := parseLogRecordKey(logRecord.Key)
			logRecordPos := d.index.Get(realKey)
			// 表明这条数据是最新的有效的数据
			if logRecordPos != nil && logRecordPos.Fid == dataFile.FileId && logRecordPos.Offset == offset {
				// 清除事务标记，因为数据有效无需事务序列号了
				logRecord.Key = logRecordKeyWithSeq(realKey, nonTransactionSeqNo)
				// 数据通过临时 DB 存储到 merge 操作对应目录数据文件下
				pos, err := mergeDB.appendLogRecord(logRecord)
				if err != nil {
					return err
				}

				if d.opt.IndexType != index.BPTree {
					// 索引信息写入 Hint 文件
					// 后续重启 DB 引擎实例，索引数据加载策略：
					// 老数据直接从索引文件加载，新数据从新数据文件加载
					if err := hintFile.WriteHintRecord(realKey, pos); err != nil {
						return err
					}
				} else {
					// 如果采用 B+ 树索引结构，需将索引信息写入 B+ 树索引文件
					// TODO: B+ 树涉及单索引文件的新老替换，目前不知道怎么解决，待完成！
					_ = mergeDB.index.Put(realKey, pos)
				}
			}

			offset += size
		}
	}

	// 持久化
	if err := hintFile.Sync(); err != nil {
		return err
	}

	if err := mergeDB.Sync(); err != nil {
		return err
	}

	// 新增标识 merge 完成文件
	mergeFinishedFile, err := data.OpenMergeFinishedFile(mergeDirPath)
	defer mergeFinishedFile.Close()
	if err != nil {
		return err
	}
	// 写入一条记录，value 记录最近未参与 merge 的文件 id
	mergeFinRecord := &data.LogRecord{
		Key:   []byte(mergeFinishedKey),
		Value: []byte(strconv.Itoa(int(nonMergeFileId))),
	}
	encRecord, _ := data.EncodeLogRecord(mergeFinRecord)
	if err := mergeFinishedFile.Write(encRecord); err != nil {
		return err
	}
	// 持久化
	if err := mergeFinishedFile.Sync(); err != nil {
		return err
	}

	return nil
}

// getMergeDirPath 获取 merge 目录路径
func (d *DB) getMergeDirPath() string {
	// 目录转换格式
	dirClean := utils.NormalizationFilePathString(d.opt.DirPath)
	// 获取上级目录
	dir := path.Dir(dirClean)
	// 获取本级目录
	base := path.Base(dirClean)
	// 拼接 上级/本级-merge
	return filepath.Join(dir, base+mergeDirName)
}

// loadMergeFiles 加载 merge 数据目录
func (d *DB) loadMergeFiles() error {
	// 获取 merge 数据目录
	mergePath := d.getMergeDirPath()
	if _, err := os.Stat(mergePath); os.IsNotExist(err) {
		return nil
	}
	defer func() {
		_ = os.RemoveAll(mergePath)
	}()

	dirEntries, err := os.ReadDir(mergePath)
	if err != nil {
		return err
	}

	// 查找标识 merge 完成文件是否存在
	var mergeFinished bool = false
	var mergeFileNames []string
	for _, entry := range dirEntries {
		if entry.Name() == data.MergeFileFinished {
			mergeFinished = true
		}
		// 事务序列号文件无需移动，因为 DB 引擎 Close 才会保存，这将不会是最新的
		if entry.Name() == data.SeqNoFileName {
			continue
		}
		// merge 时临时的 DB 引擎实例也会创建一个 B+ 树索引文件，而这个文件无用
		// TODO: 不更新merge后的索引文件，那么会有问题，待完成（索引文件的同步）！
		if entry.Name() == index.BPlusTreeFileName {
			continue
		}
		// 文件锁不用拷贝过去
		if entry.Name() == fileLockName {
			continue
		}
		mergeFileNames = append(mergeFileNames, entry.Name())
	}

	// 没有 merge 完成
	if !mergeFinished {
		return nil
	}

	nonMergeFileId, err := d.getNonMergeFileId(mergePath)
	if err != nil {
		return err
	}

	// 删除所有旧的数据文件，文件 id < nonMergeFileId
	var fileId uint32 = 0
	for ; fileId < nonMergeFileId; fileId++ {
		fileName := data.GetDataFileName(d.opt.DirPath, fileId)
		// 如果数据文件存在，直接删除

		if _, err := os.Stat(fileName); err == nil {
			if err := os.Remove(fileName); err != nil {
				return err
			}
		}
	}

	// 将新的数据文件移动到数据目录中
	for _, fileName := range mergeFileNames {
		srcPath := filepath.Join(mergePath, fileName)
		destPath := filepath.Join(d.opt.DirPath, fileName)

		if err := os.Rename(srcPath, destPath); err != nil {
			return err
		}
	}

	return nil
}

// loadIndexFromHintFile 从 Hint 文件中加载数据索引项
func (d *DB) loadIndexFromHintFile() error {
	// 查看 hint 索引文件是否存在
	hintFileName := filepath.Join(d.opt.DirPath, data.HintFileName)
	if _, err := os.Stat(hintFileName); err != nil {
		return nil
	}

	// 打开 hint 索引文件，根据配置获取加载时文件的 IO 类型
	var ioType fio.FileIOType = fio.StandardFIO
	if d.opt.MMapAtStartUp {
		ioType = fio.MemoryMap
	}
	hintFile, err := data.OpenHintFile(d.opt.DirPath, ioType)
	defer hintFile.Close()
	if err != nil {
		return err
	}

	var offset int64 = 0
	for {
		record, size, err := hintFile.ReadLogRecord(offset)
		if err != nil {
			if err == io.EOF {
				break
			}
			return err
		}

		// 解码获取位置索引信息
		recordPos := data.DecodeLogRecordPos(record.Value)
		d.index.Put(record.Key, recordPos)

		offset += size
	}
	return nil
}

// getNonMergeFileId 获取最近没参与 merge 文件的 id
func (d *DB) getNonMergeFileId(dirPath string) (uint32, error) {
	dataFile, err := data.OpenMergeFinishedFile(dirPath)
	defer dataFile.Close()
	if err != nil {
		return 0, err
	}
	record, _, err := dataFile.ReadLogRecord(0)
	if err != nil {
		return 0, err
	}
	nonMergeFileId, err := strconv.Atoi(string(record.Value))
	return uint32(nonMergeFileId), nil
}
